package com.mqstudy.rabbitmq.consumer;

import com.mqstudy.common.model.Message;
import com.mqstudy.rabbitmq.config.RabbitMQConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message as AmqpMessage;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import com.rabbitmq.client.Channel;

import java.io.IOException;

/**
 * RabbitMQ消费者
 */
@Slf4j
@Component
public class RabbitMQConsumer {
    
    /**
     * 直连队列消费者
     */
    @RabbitListener(queues = RabbitMQConfig.DIRECT_QUEUE)
    public void handleDirectMessage(Message message, Channel channel, 
                                   @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        try {
            log.info("收到直连消息: {}", message);
            
            // 模拟业务处理
            processMessage(message);
            
            // 手动确认消息
            channel.basicAck(deliveryTag, false);
            log.info("直连消息处理完成，已确认");
            
        } catch (Exception e) {
            log.error("处理直连消息失败: {}", e.getMessage(), e);
            try {
                // 拒绝消息，不重新入队
                channel.basicNack(deliveryTag, false, false);
            } catch (IOException ex) {
                log.error("拒绝消息失败: {}", ex.getMessage(), ex);
            }
        }
    }
    
    /**
     * 扇出队列1消费者
     */
    @RabbitListener(queues = RabbitMQConfig.FANOUT_QUEUE_1)
    public void handleFanoutMessage1(Message message, Channel channel,
                                    @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        try {
            log.info("扇出队列1收到消息: {}", message);
            processMessage(message);
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            log.error("处理扇出消息1失败: {}", e.getMessage(), e);
            try {
                channel.basicNack(deliveryTag, false, false);
            } catch (IOException ex) {
                log.error("拒绝扇出消息1失败: {}", ex.getMessage(), ex);
            }
        }
    }
    
    /**
     * 扇出队列2消费者
     */
    @RabbitListener(queues = RabbitMQConfig.FANOUT_QUEUE_2)
    public void handleFanoutMessage2(Message message, Channel channel,
                                    @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        try {
            log.info("扇出队列2收到消息: {}", message);
            processMessage(message);
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            log.error("处理扇出消息2失败: {}", e.getMessage(), e);
            try {
                channel.basicNack(deliveryTag, false, false);
            } catch (IOException ex) {
                log.error("拒绝扇出消息2失败: {}", ex.getMessage(), ex);
            }
        }
    }
    
    /**
     * 主题队列1消费者
     */
    @RabbitListener(queues = RabbitMQConfig.TOPIC_QUEUE_1)
    public void handleTopicMessage1(Message message, Channel channel,
                                   @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        try {
            log.info("主题队列1收到消息: {}", message);
            processMessage(message);
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            log.error("处理主题消息1失败: {}", e.getMessage(), e);
            try {
                channel.basicNack(deliveryTag, false, false);
            } catch (IOException ex) {
                log.error("拒绝主题消息1失败: {}", ex.getMessage(), ex);
            }
        }
    }
    
    /**
     * 主题队列2消费者
     */
    @RabbitListener(queues = RabbitMQConfig.TOPIC_QUEUE_2)
    public void handleTopicMessage2(Message message, Channel channel,
                                   @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        try {
            log.info("主题队列2收到消息: {}", message);
            processMessage(message);
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            log.error("处理主题消息2失败: {}", e.getMessage(), e);
            try {
                channel.basicNack(deliveryTag, false, false);
            } catch (IOException ex) {
                log.error("拒绝主题消息2失败: {}", ex.getMessage(), ex);
            }
        }
    }
    
    /**
     * 死信队列消费者
     */
    @RabbitListener(queues = RabbitMQConfig.DEAD_LETTER_QUEUE)
    public void handleDeadLetterMessage(Message message, Channel channel,
                                       @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        try {
            log.warn("收到死信消息: {}", message);
            
            // 死信消息的特殊处理逻辑
            processDeadLetterMessage(message);
            
            channel.basicAck(deliveryTag, false);
            log.info("死信消息处理完成");
            
        } catch (Exception e) {
            log.error("处理死信消息失败: {}", e.getMessage(), e);
            try {
                channel.basicNack(deliveryTag, false, false);
            } catch (IOException ex) {
                log.error("拒绝死信消息失败: {}", ex.getMessage(), ex);
            }
        }
    }
    
    /**
     * 处理消息的通用方法
     */
    private void processMessage(Message message) {
        log.info("开始处理消息: {}", message.getMessageId());
        
        // 模拟业务处理时间
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        log.info("消息处理完成: {}", message.getMessageId());
    }
    
    /**
     * 处理死信消息
     */
    private void processDeadLetterMessage(Message message) {
        log.warn("处理死信消息: {}", message.getMessageId());
        
        // 可以在这里实现死信消息的特殊处理逻辑
        // 比如：记录日志、发送告警、重试等
        
        try {
            Thread.sleep(200);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        log.warn("死信消息处理完成: {}", message.getMessageId());
    }
} 